package com.nulldev.util.concurrency.threadIt.v4.sched.impl.CAS;

import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import com.nulldev.util.VariableAPI.MathUtil;
import com.nulldev.util.VariableAPI.util.random.GlobalRandom;
import com.nulldev.util.concurrency.threadIt.v4.emapi.ExecutorManager;
import com.nulldev.util.concurrency.threadIt.v4.emapi.IExecutorManager;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExecutorFeatureExtensions;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.ExtendedExecutor;
import com.nulldev.util.concurrency.threadIt.v4.emapi.extensions.IFutureTaskFlags;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Executable;
import com.nulldev.util.concurrency.threadIt.v4.emapi.uft.Returnable;
import com.nulldev.util.concurrency.threadIt.v4.sched.ASyncScheduler;
import com.nulldev.util.data.Variables;
import com.nulldev.util.internal.backport.concurrency9.Lists;

public class ContextuallyAwareAsyncScheduler implements ASyncScheduler, IFutureTaskFlags {

	private final LinkedBlockingDeque<Runnable> QUEUE;
	private static final boolean ENABLE_PRIORITY_SORTING = true;
	private static final int REDUCED_QUEUE_SIZE = 48;

	private final IExecutorManager execMan;
	private final AtomicBoolean allowL1Batching = new AtomicBoolean(ExecutorManager.ENABLE_BATCH_OPERATIONS),
			allowL2Batching = new AtomicBoolean(ExecutorManager.ENABLE_BATCH_OPERATIONS),
			allowL1Optimizations = new AtomicBoolean(ExecutorManager.ENABLE_BATCH_OPERATIONS && ENABLE_PRIORITY_SORTING),
			executorSupportsBatching = new AtomicBoolean(), isExtended = new AtomicBoolean();

	public ContextuallyAwareAsyncScheduler(final IExecutorManager execMan) {
		Variables.requireNonNull(execMan, "execMan");
		this.execMan = execMan;
		if (this.execMan.executor() instanceof ExtendedExecutor
				&& ((ExtendedExecutor) this.execMan.executor()).supportsFeature(ExecutorFeatureExtensions.EEX_executor_sched_accepts_always)) {
			this.QUEUE = new LinkedBlockingDeque<Runnable>(REDUCED_QUEUE_SIZE);
		} else {
			this.QUEUE = new LinkedBlockingDeque<Runnable>(272);
		}
	}

	@Override
	public void executorUpdated(final ExecutorService ex) {
		Variables.requireNonNull(ex, "ex");
		this.isExtended.set(this.execMan.executor() instanceof ExtendedExecutor);
		this.allowL1Batching.set(this.execMan.supportsNativeBatching() && !this.execMan.isWorkingAroundBatching()
				&& !this.execMan.isWorkingAroundLevelledBatching() && ExecutorManager.ENABLE_BATCH_OPERATIONS && this.isExtended.get());
		this.allowL2Batching.set(this.execMan.supportsNativeBatching() && this.execMan.supportsTaskDrain() && !this.execMan.isWorkingAroundLevelledBatching()
				&& !this.execMan.isWorkingAroundBatching() && ExecutorManager.ENABLE_BATCH_OPERATIONS && this.isExtended.get());
		this.allowL1Optimizations.set(this.execMan.supportsNativeBatching() && this.execMan.supportsTaskDrain() && this.execMan.supportsSchedQueueBypass()
				&& !this.execMan.isWorkingAroundBatching() && ExecutorManager.ENABLE_BATCH_OPERATIONS && this.isExtended.get());
		this.executorSupportsBatching.set(this.execMan.supportsNativeBatching());
	}

	@Override
	public void run() {
		this.isExtended.set(this.execMan.executor() instanceof ExtendedExecutor);
		this.allowL1Batching.set(this.execMan.supportsNativeBatching() && !this.execMan.isWorkingAroundLevelledBatching()
				&& !this.execMan.isWorkingAroundBatching() && ExecutorManager.ENABLE_BATCH_OPERATIONS && this.isExtended.get());
		this.allowL2Batching.set(this.execMan.supportsNativeBatching() && this.execMan.supportsTaskDrain() && !this.execMan.isWorkingAroundLevelledBatching()
				&& !this.execMan.isWorkingAroundBatching() && ExecutorManager.ENABLE_BATCH_OPERATIONS && this.isExtended.get());
		this.allowL1Optimizations.set(this.execMan.supportsNativeBatching() && this.execMan.supportsTaskDrain() && this.execMan.supportsSchedQueueBypass()
				&& !this.execMan.isWorkingAroundBatching() && ExecutorManager.ENABLE_BATCH_OPERATIONS && this.isExtended.get());
		this.executorSupportsBatching.set(this.execMan.supportsNativeBatching());
		while (this.execMan.isRunning()) {
			try {
				final int SIZE = this.QUEUE.size();
				final boolean canBatch = SIZE >= BATCH_SIZE;
				final Runnable x;
				try {
					x = this.QUEUE.take();
				} catch (InterruptedException e) {
					continue;
				}
				if (x == null)
					continue;
				else if (canBatch && this.executorSupportsBatching.get() && this.allowL2Batching.get()) {
					final int batchSize = MathUtil.clamp(1, MAXIMUM_BATCH_SIZE, SIZE - 1);
					((ExtendedExecutor) this.execMan.executor()).batch(x);
					((ExtendedExecutor) this.execMan.executor()).executor_drainTasks(this.QUEUE, batchSize);
				} else if (this.allowL1Batching.get() && this.executorSupportsBatching.get()) {
					((ExtendedExecutor) this.execMan.executor()).batch(x);
				} else {
					this.execMan.executor().execute(x);
				}
			} catch (Throwable t) {
				t.printStackTrace();
			}
		}
	}

	@Override
	public void submit_async(final Runnable r) {
		if (r == null)
			return;
		if (!this.QUEUE.offerLast(r))
			try {
				this.QUEUE.put(r);
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}
	}

	@Override
	public <T> Executable<T> submit_async(final Executable<T> r) {
		if (r == null)
			throw new NullPointerException("r == null!");
		if (r instanceof Runnable) {
			if (offerToQueue(r)) {
				r.signalQueued();
			} else {
				r.rejectSubmission();
			}
			return r;
		} else {
			System.err.println("[CAAS::submit_async] Attempted to submit an asynchronous task using the deprecated logic!!!");
			try {
				this.QUEUE.put(() -> {
					r.run_st();
				});
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			} finally {
				r.signalQueued();
			}
			return r;
		}
	}

	private <T> boolean attemptOfferToQueue(final Executable<T> f) {
		try {
			return this.QUEUE.offerFirst((Runnable) f, 333, TimeUnit.MILLISECONDS);
		} catch (InterruptedException e) {
			return false;
		}
	}

	private <T> boolean offerToQueue(final Executable<T> f) {
		final boolean isHighPriority = (int) f.getFlag(FT_HIGH_PRIORITY) == FT_TRUE || (int) f.getFlag(FT_PREFER_LOW_LATENCY) == FT_TRUE;
		final boolean isAttemptable = (int) f.getFlag(FT_ATTEMPTABLE_TASK) == FT_TRUE;
		/* L1 optimization: Bypass queuing to reach Executor quicker. */
		if (isHighPriority && !isAttemptable && this.allowL1Optimizations.get()) {
			this.execMan.executor().execute((Runnable) f);
			return true;
		} else {
			return (isHighPriority ? this.QUEUE.offerFirst((Runnable) f) : this.QUEUE.offerLast((Runnable) f));
		}
	}

	@Override
	public <T> Executable<T> submit_async(final Returnable<T> r) {
		if (r == null)
			throw new NullPointerException("r == null!");
		try {
			final Executable<T> f = IExecutorManager.makeTaskNoAccel(r, IFutureTaskFlags.FT_PREFER_BATCHING);
			if (!offerToQueue(f)) {
				f.rejectSubmission();
			} else {
				f.signalQueued();
			}
			return f;
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	@Override
	public <T> boolean attempt_async_submit(Executable<T> r) {
		if (r == null)
			throw new NullPointerException("r == null!");
		try {
			if (this.attemptOfferToQueue(r)) {
				r.signalQueued();
				return true;
			} else {
				r.rejectSubmission();
				return false;
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	private static final Comparator<Runnable> PRIORITY_SORTER = new Comparator<Runnable>() {

		@Override
		public int compare(final Runnable o1, final Runnable o2) {
			if (o1 == null || o2 == null)
				return 0;
			if (!(o1 instanceof Executable<?>) || !(o2 instanceof Executable<?>))
				return 0;
			if ((int) ((Executable<?>) o1).getFlag(FT_HIGH_PRIORITY) == FT_TRUE)
				return -1;
			if ((int) ((Executable<?>) o2).getFlag(FT_HIGH_PRIORITY) == FT_TRUE)
				return 1;
			if ((int) ((Executable<?>) o1).getFlag(FT_PREFER_LOW_LATENCY) == FT_TRUE)
				return -1;
			if ((int) ((Executable<?>) o2).getFlag(FT_PREFER_LOW_LATENCY) == FT_TRUE)
				return 1;
			return 0;
		}
	};

	private void submit_all_async_batch(final Runnable... r) {
		final List<Runnable> s = Arrays.asList(r);
		if (ENABLE_PRIORITY_SORTING) {
			Collections.sort(s, PRIORITY_SORTER);
		} else {
			Collections.shuffle(s, GlobalRandom.getFallbackRandom());
		}
		s.forEach(t -> {
			try {
				QUEUE.put(t);
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			}
		});
	}

	@Override
	public void submit_all_async(final Runnable... r) {
		if (r == null || r.length == 0) {
			return;
		} else if (r.length > REDUCED_QUEUE_SIZE) {
			final int safeToFit = REDUCED_QUEUE_SIZE - this.QUEUE.size();
			if (safeToFit <= 0)
				throw new ArrayStoreException("submit_all_async: Queue is full!");
			final Runnable[] safe = new Runnable[safeToFit];
			System.arraycopy(r, 0, safe, 0, safeToFit);
			final Runnable[] late = new Runnable[r.length - safeToFit];
			System.arraycopy(r, safeToFit, late, 0, (r.length - safeToFit));
			if (this.allowL1Optimizations.get()) {
				submit_all_async_batch(safe);
				/*
				 * L1 Optimization: Don't bother submitting now, also allow for more fine grain
				 * batching
				 */
				submit_all_async(late);
			} else {
				submit_all_async_batch(safe);
				submit_all_async_batch(late);
			}
		} else if (r.length > 8) {
			submit_all_async_batch(r);
		} else {
			this.QUEUE.addAll(Lists.of(r));
		}
	}

	@Override
	public void clearQueues() {
		this.QUEUE.clear();
	}

	@Override
	public boolean supportsDebugQuery() {
		return true;
	}

	@Override
	public Object debug_query(final ASyncSchedulerDebugQuery query) {
		switch (Variables.requireNonNullObject(query, "query")) {
			case ASYNC_SCHED_DEBUG_QUEUE_SIZE:
				return this.QUEUE.size();
			case ASYNC_SCHED_DEBUG_L1_BATCHING_ENABLED:
				return this.allowL1Batching.get();
			case ASYNC_SCHED_DEBUG_L2_BATCHING_ENABLED:
				return this.allowL2Batching.get();
			case ASYNC_SCHED_DEBUG_L1_OPTIMIZATIONS_ENABLED:
				return this.allowL1Optimizations.get();
			case ASYNC_SCHED_DEBUG_EXECUTOR_BATCHING_SUPPORTED:
				return this.executorSupportsBatching.get();
			default:
				return null;
		}
	}

}
